package com.huan.orderpay_detect;

import com.huan.beans.OrderEvent;
import com.huan.beans.ReceiptEvent;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class TxPayMatchByJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism( 1 );
        env.setStreamTimeCharacteristic( TimeCharacteristic.EventTime );

        // 读取数据并转换成POJO类型
        // 读取订单支付事件数据
        String filePath = "E:\\Project\\UserBehaviorAnalysis_Java\\OrderTimeoutDetect\\src\\main\\resources\\OrderLog.csv";
        DataStream<OrderEvent> orderEventStream = env.readTextFile( filePath )
                .map( line -> {
                    String[] fields = line.split( "," );
                    return new OrderEvent( new Long( fields[0] ), fields[1], fields[2], new Long( fields[3] ) );
                } )
                .assignTimestampsAndWatermarks( new AscendingTimestampExtractor<OrderEvent>() {
                    @Override
                    public long extractAscendingTimestamp(OrderEvent element) {
                        return element.getTimestamp() * 1000L;
                    }
                } )
                .filter( data -> !"".equals( data.getTxId() ) ); // 交易ID不能为空

        //读取到账事件数据
        String path = "E:\\Project\\UserBehaviorAnalysis_Java\\OrderTimeoutDetect\\src\\main\\resources\\ReceiptLog.csv";
        SingleOutputStreamOperator<ReceiptEvent> receiptEventStream = env.readTextFile( path ).map( line -> {
            String[] fields = line.split( "," );
            return new ReceiptEvent( fields[0], fields[1], new Long( fields[2] ) );
        } )
                .assignTimestampsAndWatermarks( new AscendingTimestampExtractor<ReceiptEvent>() {
                    @Override
                    public long extractAscendingTimestamp(ReceiptEvent element) {
                        return element.getTimestamp() * 1000L;
                    }
                } );

        //区间连接两条流，得到匹配的数据
        SingleOutputStreamOperator<Tuple2<OrderEvent,ReceiptEvent>> resultStream = orderEventStream.keyBy( OrderEvent::getTxId )
                .intervalJoin( receiptEventStream.keyBy( ReceiptEvent::getTxId ) )
                // -3,5 区间范围
                .between( Time.seconds( -3 ), Time.seconds( 5 ) )
                .process( new TxPayMatchDetectByJoin() );

        resultStream.print();

        env.execute("TxPayMatchByJoin");
    }

    //实现自定义ProcessJoinFunction
    public static class TxPayMatchDetectByJoin extends ProcessJoinFunction<OrderEvent,ReceiptEvent, Tuple2<OrderEvent,ReceiptEvent>> {


        @Override
        public void processElement(OrderEvent left, ReceiptEvent right, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            out.collect( new Tuple2<>( left, right ) );

        }
    }
}
